package com.atguigu.gmall.realtime.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

/**
 * @author Felix
 * @date 2024/7/26
 * 将流中数据同步到HBase
 */
public class DimSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {
    private Connection hBaseConn;
    private Jedis jedis;
    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseConn = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConnection(hBaseConn);
        RedisUtil.closeJedis(jedis);
    }

    //将流中数据写到HBase表中
    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> tup2, Context context) throws Exception {
        //({"tm_name":"Redmi","id":1,"type":"update"},TableProcessDim(sourceTable=base_trademark, sinkTable=dim_base_trademark, sinkColumns=id,tm_name, sinkFamily=info, sinkRowKey=id, op=r))
        JSONObject jsonObj = tup2.f0;
        TableProcessDim tableProcessDim = tup2.f1;
        //获取对业务数据库维度表操作的类型
        String type = jsonObj.getString("type");
        jsonObj.remove("type");

        //获取操作的HBase表的名
        String sinkTable = tableProcessDim.getSinkTable();
        //获取主键
        String rowKey = jsonObj.getString(tableProcessDim.getSinkRowKey());
        if("delete".equals(type)){
            //说明对业务数据库维度表进行了删除操作，需要在HBase对应的表中也执行删除
            HBaseUtil.delRow(hBaseConn, Constant.HBASE_NAMESPACE,sinkTable,rowKey);
        }else{
            //说明对业务数据库维度表进行了添加或者修改，需要在HBase对应的表中执行put操作
            String family = tableProcessDim.getSinkFamily().split(",")[0];
            HBaseUtil.putRow(hBaseConn,Constant.HBASE_NAMESPACE,sinkTable,rowKey,family,jsonObj);
        }

        //如果维度数据  发生了变化，将Redis中缓存的维度数据清除掉
        if("update".equals(type)||"delete".equals(type)){
            String key = RedisUtil.getKey(sinkTable, rowKey);
            jedis.del(key);
        }

    }
}
